package com.navinfo.tripanalysis.service.impl;

import com.navinfo.tripanalysis.pojo.LoadDataParam;
import com.navinfo.tripanalysis.pojo.Point;
import com.navinfo.tripanalysis.service.AbstractLoad0F37DataServiceBigData;
import com.navinfo.tripanalysis.util.BigDataUtils;
import com.navinfo.tripanalysis.util.CommonUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.SparkSession;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;

/**
 * 通过Hive加载0f37数据
 * @author 沈东生
 */
public class Load0F37DataServiceHiveImpl extends AbstractLoad0F37DataServiceBigData {
    private static final Logger logger = LoggerFactory.getLogger(Load0F37DataServiceHiveImpl.class);
    private String dbName;
    private String tableName;

    public String getDbName() {
        return dbName;
    }
    public void setDbName(String dbName) {
        this.dbName = dbName;
    }
    public String getTableName() {
        return tableName;
    }
    public void setTableName(String tableName) {
        this.tableName = tableName;
    }

    @Override
    public JavaRDD<Point> load(SparkSession spark, JavaSparkContext jsc, LoadDataParam param) {
        List<Integer> batchList = (List<Integer>) param.getExtendParam("batchList");
        StringBuilder sb = new StringBuilder();
        sb.append(String.format("select tid, %s ", CommonUtils.mkString(COL_NAMES,",")));
        sb.append("\n");
        sb.append(" from ");
        if(StringUtils.isNotEmpty(getDbName())){
            sb.append(getDbName()).append(".");
        }
        sb.append(getTableName());
        sb.append("\n");
        sb.append(" where ");
        sb.append(BigDataUtils.getBatchWhereSql(param.getDay(), batchList, param.getTids()));
        sb.append(" and ").append(getWhereSql0f37(param));

        String sql = sb.toString();
        logger.error("load 0f37 sql:-----------------\n{}", sql);
        return toJavaRdd(spark.sql(sql));
    }
}
